package com.zzl.kafkatest;

import kafka.producer.KeyedMessage;
import kafka.producer.Producer;
import kafka.producer.ProducerConfig;
import scala.collection.JavaConverters;
import scala.collection.Seq;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;

public class MyProducer {
    //kafka创建的topic
    public static final String TOPIC = "zzl_test";
    //要发送的内容
    private static final String CONTENT = "This is a single message";
    //broker的地址和端口
    private static final String BROKER_LIST = "172.31.0.91:9092";
    // 序列化类
    private static final String SERIALIZER_CLASS = "kafka.serializer.StringEncoder";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("serializer.class", SERIALIZER_CLASS);
        props.put("metadata.broker.list", BROKER_LIST);


        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> producer = new Producer<String, String>(config);

        //Send one message.
//        KeyedMessage<String, String> message = new KeyedMessage<String, String>(TOPIC, CONTENT);
//
//        producer.send((Seq<KeyedMessage<String, String>>) message);

        //Send multiple messages.
        List<KeyedMessage<String,String>> messages = new ArrayList<KeyedMessage<String, String>>();
        for (int i = 0; i < 5; i++) {
            messages.add(new KeyedMessage<String, String>(TOPIC, "Multiple message at a time. " + i));
        }
        Seq<KeyedMessage<String, String>> keyedMessageSeq = JavaConverters.asScalaIteratorConverter(messages.iterator()).asScala().toSeq();
        producer.send(keyedMessageSeq);

        producer.close();
    }

}
